package udf.matching;

import java.io.IOException;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

public class SetDifferenceSimple extends EvalFunc<DataBag> {

	private static  boolean initialized = false;
	private static HashSet<String> seeds= new HashSet<String>();
	
	private BagFactory bagFactory = BagFactory.getInstance();

/**
 * Tuple has the following fields
 * 
 * 
 * 0  -  set
 * 
 * 1  -  seed tags
 * @throws ExecException 
 * 
 * 
 * 
 */
		
	public void init(Tuple input) throws ExecException{
		
		DataBag bag= (DataBag) input.get(2);
		
		
		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();
			String tag = (String)tup.get(0);
			seeds.add(tag);
			
				
			}
		
	} 
	

	public DataBag exec(Tuple input) throws IOException {
			
			DataBag set = (DataBag) input.get(1);
			
			
			Tuple tt = (Tuple) input.get(0);
			
			String query =(String) tt.get(0);
			String t_tag =(String) tt.get(1);
			String candidate =(String) tt.get(2);
			
			
			Integer pos_tag = (Integer) input.get(3);
			
			HashSet<String> s= new HashSet<String>();
			
			
			if(initialized==false){
				
			init(input);
			initialized=true;
			}
			
			
			
			
			DataBag output = bagFactory.newDefaultBag();

			Tuple t = TupleFactory.getInstance().newTuple();
			
			 Iterator<String> iterator = seeds.iterator();
			 
			
			Iterator<Tuple> set_iter = set.iterator();
			HashSet<String> t_original = new HashSet<String>();
			while(set_iter.hasNext()){
				Tuple aa = set_iter.next();
				
				t_original.add((String)aa.get(pos_tag));
			}
			 		 
			while(iterator.hasNext()){
				 
				 String temporal = iterator.next();
				 
				 
				 if(!t_original.contains(temporal)){
					 
					 s.add(temporal);
				 }
				 
			 }
			
			
			Iterator<String>iter = s.iterator();
			while(iter.hasNext()){
				t = TupleFactory.getInstance().newTuple();
				t.append(query);
				t.append(t_tag);
				t.append(candidate);
				t.append(iter.next());
				output.add(t);
				
			}
			
//			if(output.size()==0) return null;
			
			return output;
		

	}
}

